#pragma once
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"

#include "../mqcommon/log.hpp"
#include "../mqcommon/helper.hpp"
#include "../mqcommon/threadpool.hpp"
#include "../mqcommon/msg.pb.h"
#include "../mqcommon/proto.pb.h"
#include "consumer.hpp"
#include "virtualhost.hpp"
#include "route.hpp"
#include <iostream>

namespace mymq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using OpenChannelRequestPtr = std::shared_ptr<OpenChannelRequest>;
    using CloseChannelRequestPtr = std::shared_ptr<CloseChannelRequest>;
    using DeclareExchangeRequestPtr = std::shared_ptr<DeclareExchangeRequest>;
    using DeleteExchangeRequestPtr = std::shared_ptr<DeleteExchangeRequest>;
    using DeclareQueueRequestPtr = std::shared_ptr<DeclareQueueRequest>;
    using DeleteQueueRequestPtr = std::shared_ptr<DeleteQueueRequest>;
    using QueueBindRequestPtr = std::shared_ptr<QueueBindRequest>;
    using QueueUnBindRequestPtr = std::shared_ptr<QueueUnBindRequest>;
    using PublishRequestPtr = std::shared_ptr<PublishRequest>;
    using AckRequestPtr = std::shared_ptr<AckRequest>;
    using ConsumeRequestPtr = std::shared_ptr<ConsumeRequest>;
    using CancelRequestPtr = std::shared_ptr<CancelRequest>;
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
    public:
        Channel(const std::string id, const ConsumerManager::ptr& cmp, const VirtualHost::ptr host, 
                const muduo::net::TcpConnectionPtr conn, const ProtobufCodecPtr& codec,
                const ThreadPool::ptr pool)
        : _id(id)
        , _consumer_manager(cmp)
        , _host(host)
        , _conn(conn)
        , _codec(codec)
        , _pool(pool)
        {
            DEBUG_LOG("new Channel:%p", this);
        }
        ~Channel()
        {
            if (_consumer.get() != nullptr)
            {
                _consumer_manager->Remove(_consumer->tag, _consumer->qname);
            }
            DEBUG_LOG("del Channel:%p", this);
        }
        void DeclareExchange(const DeclareExchangeRequestPtr &req)
        {
            bool ret = _host->DeclareExchange(req->exchange_name(), req->exchange_type(),
                                                req->durable(),
                                                req->auto_delete(), req->args());
            Response(req->rid(), req->cid(), ret);
        }
        void DeleteExchange(const DeleteExchangeRequestPtr &req)
        {
            bool ret = _host->DeleteExchange(req->exchange_name());
            Response(req->rid(), req->cid(), ret);
        }
        void DeclareQueue(const DeclareQueueRequestPtr &req)
        {
            bool ret = _host->DeclareQueue(req->queue_name(), req->durable(),
                                            req->exclusive(), req->auto_delete(),
                                            req->args());
            if (ret == false)
            {
                Response(req->rid(), req->cid(), ret);
                return;
            }
            // 初始化队列的消费者管理句柄
            _consumer_manager->InitQueueConsumer(req->queue_name());
            Response(req->rid(), req->cid(), ret);
        }
        void DeleteQueue(const DeleteQueueRequestPtr &req)
        {
            _consumer_manager->DestroyQueueConsumer(req->queue_name());
            bool ret = _host->DeleteQueue(req->queue_name());
            Response(req->rid(), req->cid(), ret);
        }
        void QueueBind(const QueueBindRequestPtr &req)
        {
            bool ret = _host->Bind(req->exchange_name(), req->queue_name(), req->binding_key());
            Response(req->rid(), req->cid(), ret);
        }
        void QueueUnBind(const QueueUnBindRequestPtr &req)
        {
            bool ret = _host->UnBind(req->exchange_name(), req->queue_name());
            Response(req->rid(), req->cid(), ret);
        }
        // 消息的发布
        void Publish(const PublishRequestPtr &req)
        {
            //1. 判断交换机是否存在
            auto ep = _host->SelectExchange(req->exchange_name());
            if (ep.get() == nullptr)
            {
                Response(req->rid(), req->cid(), false);
                return;
            }
            //2. 进行交换路由，判断消息可以发布到交换机绑定的哪个队列中
            BindingMapper::MsgQueueBindingMap mqbm = _host->GetExchangeBindings(req->exchange_name());
            BasicProperties *properties = nullptr;
            std::string routing_key;
            if (req->has_properties())
            {
                properties = req->mutable_properties();
                routing_key = properties->routing_key();
            }
            for (auto &binding : mqbm)
            {
                if (Router::Route(ep->type, routing_key, binding.second->binding_key))
                {
                    //3. 将消息添加到队列中（添加消息的管理）
                    _host->Publish(binding.first, properties, req->body());
                    //4.向线程池中添加一个消息消费任务(向指定队列的订阅者去推送消息-线程池完成)
                    auto task = std::bind(&Channel::consume, this, binding.first);
                    // _pool->Push(&Channel::consume, this, binding.first); // 不可以这样，consume是类的成员函数，必须要有对象调用才可以
                    _pool->Push(task);
                    // _pool->Push([this](const std::string& qname) { consume(qname); }, binding.first);
                }
            }
            Response(req->rid(), req->cid(), true);
        }
        void Ack(const AckRequestPtr &req)
        {
            bool ret = _host->Ack(req->queue_name(), req->message_id());
            Response(req->rid(), req->cid(), true);
        }
        // 订阅队列消息
        void Consume(const ConsumeRequestPtr &req)
        {
            bool ret = _host->ExistsQueue(req->queue_name());
            if (ret == false)
            {
                Response(req->rid(), req->cid(), false);
                return;
            }
            auto cb = std::bind(&Channel::callback, this, std::placeholders::_1,
                                std::placeholders::_2, std::placeholders::_3);
            // 创建了消费者，channel的角色就是消费者

            _consumer = _consumer_manager->Create(req->consumer_tag(), req->queue_name(),
                                                    req->auto_ack(), cb);
            Response(req->rid(), req->cid(), true);
        }
        // 取消订阅
        void Cancel(const CancelRequestPtr &req)
        {
            _consumer_manager->Remove(req->consumer_tag(), req->queue_name());
            Response(req->rid(), req->cid(), true);
        }
    private:
        void callback(const std::string& tag, const BasicProperties* bp, const std::string& body)
        {
            // 针对参数组织出推送消息，将消息推送给channel对应的客户端
            ConsumeResponse resp;
            resp.set_cid(_id);
            resp.set_consumer_tag(tag);
            resp.set_body(body);
            if (bp)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, resp);
        }
        void consume(const std::string &qname)
        {
            //指定队列消费消息
            //1. 从队列中取出一条消息
            MessagePtr mp = _host->Consume(qname); 
            if (mp.get() == nullptr)
            {
                DEBUG_LOG("执行消费任务失败，%s 队列没有消息！", qname.c_str());
                return;
            }
            //2. 从队列订阅者中取出一个订阅者
            Consumer::ptr cp = _consumer_manager->Choose(qname);
            if (cp.get() == nullptr)
            {
                DEBUG_LOG("执行消费任务失败，%s 队列没有消费者！", qname.c_str());
                return;
            }
            //3. 调用订阅者对应的消息处理函数，实现消息的推送
            cp->callback(cp->tag, mp->mutable_payload()->mutable_properties(),
                            mp->payload().body());
            //4. 判断如果订阅者是自动确认---不需要等待确认，直接删除消息,
            //   否则需要外部收到消息确认后再删除
            if (cp->auto_ack)
                _host->Ack(qname, mp->payload().properties().id());
        }
        void Response(const std::string rid, const std::string cid, bool ok)
        {
            CommonResponse resp;
            resp.set_rid(rid);
            resp.set_cid(cid);
            resp.set_ok(ok);
            _codec->send(_conn, resp);
        }
    private:
        std::string _id;
        Consumer::ptr _consumer;
        ConsumerManager::ptr _consumer_manager;
        VirtualHost::ptr _host;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        ThreadPool::ptr _pool;
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
    public:
        ChannelManager()
        {
        }
        bool CreateChannel(const std::string id, const ConsumerManager::ptr& cmp, const VirtualHost::ptr host, 
                const muduo::net::TcpConnectionPtr conn, const ProtobufCodecPtr& codec,
                const ThreadPool::ptr pool)
        {
            std::lock_guard<std::mutex> lock(_mtx);
            auto it = _channels.find(id);
            if (it != _channels.end())
            {
                INFO_LOG("信道：%s 已经存在,无须再创建", id.c_str());
                return false;
            }
            auto channel = std::make_shared<Channel>(id, cmp, host, conn, codec, pool);
            _channels.insert(std::make_pair(id, channel));
            return true;
        }
        bool DestoryChannel(const std::string id)
        {
            std::lock_guard<std::mutex> lock(_mtx);
            auto it = _channels.find(id);
            if (it == _channels.end())
            {
                INFO_LOG("信道：%s 不存在,无须销毁", id.c_str());
                return false;
            }
            _channels.erase(id);
            return true;
        }
        Channel::ptr SelectChannel(const std::string id)
        {
            std::lock_guard<std::mutex> lock(_mtx);
            auto it = _channels.find(id);
            if (it == _channels.end())
            {
                INFO_LOG("信道：%s 不存在,返回一个空信道", id.c_str());
                return Channel::ptr();
            }
            return it->second;
        }
    private:
        std::mutex _mtx;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}